# Copyright (c) HySoP 2011-2024
#
# This file is part of HySoP software.
# See "https://particle_methods.gricad-pages.univ-grenoble-alpes.fr/hysop-doc/"
# for further info.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import inspect
from abc import ABCMeta, abstractmethod
from hysop.constants import Implementation, Backend, implementation_to_backend
from hysop.tools.decorators import debug
from hysop.tools.htypes import check_instance, first_not_None
from hysop.core.graph.node_generator import ComputationalGraphNodeGenerator
from hysop.fields.continuous_field import Field
from hysop.topology.topology import Topology
[docs]
class ComputationalGraphNodeFrontend(ComputationalGraphNodeGenerator):
@debug
def __new__(
cls,
implementation=None,
base_kwds=None,
candidate_input_tensors=None,
candidate_output_tensors=None,
**impl_kwds,
):
base_kwds = {} if (base_kwds is None) else base_kwds
return super().__new__(
cls,
candidate_input_tensors=candidate_input_tensors,
candidate_output_tensors=candidate_output_tensors,
**base_kwds,
)
@debug
def __init__(
self,
implementation=None,
base_kwds=None,
candidate_input_tensors=None,
candidate_output_tensors=None,
**impl_kwds,
):
"""
Initialize a ComputationalGraphNodeFrontend
Parameters
----------
implementation: Implementation, optional, defaults to None
target implementation, should be contained in available_implementations().
If None, implementation will be set to default_implementation().
base_kwds: dict, optional, defaults to None
Base class keywords arguments.
If None, an empty dict will be passed.
impl_kwds:
Keywords arguments that will be passed towards implementation implemention
__init__.
Attributes
----------
implementation: Implementation
the implementation key
backend: Backend
the backend corresponding to implementation
impl: ComputationalGraphNodeGenerator or ComputationalGraphNode
the implementation class
impl_kwds:
Keywords arguments that will be passed towards implementation implemention
impl.__init__ during a call to _generate.
"""
base_kwds = first_not_None(base_kwds, {})
candidate_input_tensors = first_not_None(candidate_input_tensors, ())
candidate_output_tensors = first_not_None(candidate_output_tensors, ())
if "variables" in impl_kwds:
variables = impl_kwds["variables"]
candidate_input_tensors += tuple(variables.keys())
candidate_output_tensors += tuple(variables.keys())
super().__init__(
candidate_input_tensors=candidate_input_tensors,
candidate_output_tensors=candidate_output_tensors,
**base_kwds,
)
check_instance(implementation, Implementation, allow_none=True)
default_implementation = self.default_implementation()
available_implementations = self.available_implementations()
if not isinstance(default_implementation, Implementation):
msg = "default_implementation is not a instance of hysop.backend.Implementation."
raise TypeError(msg)
for b in available_implementations:
if not isinstance(b, Implementation):
msg = "{} is not a instance of hysop.backend.Implementation."
msg = msg.format(b)
raise TypeError(msg)
if default_implementation not in available_implementations:
msg = (
"default_implementation is not contained in available_implementations."
)
raise ValueError(msg)
if implementation is None:
implementation = default_implementation
elif implementation not in available_implementations:
simplementations = []
simplementations.append(f"-{default_implementation} (default)")
for b in available_implementations:
if b != default_implementation:
simplementations.append(f"-{b}")
msg = "Specified implementation '{}' is not an available implementation, "
msg += "available implementations are:\n {}"
msg = msg.format(implementation, "\n ".join(simplementations))
raise ValueError(msg)
elif self.implementations()[implementation] is None:
msg = "Specified implementation '{}' is registered as an available implementation for operator '{}', "
msg += "but no underlying implementation was found. This may be due to missing dependency or a catched "
msg += "import error in file file://{}."
msg = msg.format(
implementation,
self.__class__.__name__,
inspect.getfile(self.__class__)[:-1],
)
raise ValueError(msg)
self.implementation = implementation
self.backend = implementation_to_backend(implementation)
self.impl = self.implementations()[implementation]
self.impl_kwds = impl_kwds
self._generated = False
self._input_fields_to_dump = []
self._output_fields_to_dump = []
@debug
def _generate(self):
try:
op = self.impl(**self.impl_kwds)
except:
sargs = [f'*{k} = {repr(v)}'
for (k, v) in self.impl_kwds.items()]
msg = f'FATAL ERROR during {self.__class__}.generate():\n'
msg += f' => failed to initialize an instance of type {self.impl}'
msg += '\n by using the following keyword arguments:'
msg += '\n '+'\n '.join(sargs)
print(f'\n{msg}')
raise
for kwds in self._input_fields_to_dump:
op.dump_inputs(**kwds)
for kwds in self._output_fields_to_dump:
op.dump_outputs(**kwds)
self._generated = True
return (op,)
[docs]
@abstractmethod
def implementations(cls):
"""
Should return all implementations as a dictionnary.
Keys are Implementation instances and values are either ComputationalGraphNode
or ComputationalGraphNodeGenerator.
"""
pass
[docs]
@abstractmethod
def default_implementation(cls):
"""
Return the default Implementation, should be compatible with available_implementations.
"""
pass
[docs]
@classmethod
def available_implementations(cls):
"""
Return all available implementations.
"""
return cls.implementations().keys()
[docs]
def dump_outputs(self, **kwds):
"""
Tell the generated operator to dump some of its outputs after
apply is called.
Target folder, file, dump frequency and other io pameters
are passed trough instance io_params of this parameter or
as keywords.
See hysop.core.computational_node.ComputationalGraphNode.dump_outputs().
"""
msg = "Cannot dump outputs after {} has been generated."
msg = msg.format(self.name)
assert self._generated is False, msg
self._output_fields_to_dump.append(kwds)
[docs]
class MultiComputationalGraphNodeFrontend(ComputationalGraphNodeFrontend):
"""
Interface like ComputationalGraphNodeFrontend that handles multiple
different discretization methods: IMPLEMENTATION_KEY + IMPLEMENTATION => OPERATOR
(ComputationalGraphNodeFrontend only generates OPERATOR from IMPLEMENTATION).
"""
@debug
def __new__(cls, implementation_key, implementation=None, **kwds):
return super().__new__(cls, implementation=implementation, **kwds)
@debug
def __init__(self, implementation_key, implementation=None, **kwds):
"""
Initialize a MultiComputationalGraphNodeFrontend
Parameters
----------
implementation_key: object
Target implementation method.
Should be contained in available_implementation_keys().
implementation: Implementation, optional, defaults to None
Target implementation, should be contained in available_implementations()[implementation_key].
If None, implementation will be set to default_implementation().
kwds: dict
Base class keyword arguments.
Attributes
----------
implementation_key: object
Target implementation method.
"""
if implementation_key not in self.available_implementation_keys():
keys = []
for k in self.available_implementation_keys():
keys.append(f"-{k}")
msg = "Specified implementation method '{}' is not an available implementation, "
msg += "available implementations methods are:\n {}"
msg = msg.format(implementation_key, "\n ".join(keys))
raise ValueError(msg)
self._impl_key = implementation_key
super().__init__(implementation=implementation, **kwds)
[docs]
@abstractmethod
def all_implementations(cls):
"""
Return all implementations of a certain method as a dictionnary.
Keys are the methods and values are dictionnaries of (implementation -> operator).
"""
pass
[docs]
@abstractmethod
def all_default_implementations(cls):
"""
Return the default Implementation for each method.
The return type is a dictionnary (method -> default_implementation).
"""
pass
[docs]
@classmethod
def available_implementation_keys(cls):
"""
Return all available implementation methods.
"""
return cls.all_implementations().keys()
[docs]
def implementations(self):
return self.all_implementations()[self._impl_key]
[docs]
def available_implementations(self):
return self.implementations().keys()
[docs]
def default_implementation(self):
return self.all_default_implementations()[self._impl_key]
@property
def implementation_key(self):
return self._impl_key
